package bsf.baseservice.businessmq.producter;

import bsf.baseservice.businessmq.common.BusinessMQConfig;
import bsf.baseservice.businessmq.dal.tb_messagequeue_dal;
import bsf.baseservice.businessmq.db.ProducterBLL;
import bsf.baseservice.businessmq.model.tb_messagequeue_model;
import bsf.baseservice.businessmq.redis.RedisNetCommand;
import bsf.baseservice.businessmq.systemruntime.log.DebugHelper;
import bsf.baseservice.businessmq.systemruntime.log.LogHelper;
import bsf.baseservice.businessmq.producter.loadbalance.BaseLoadBalance;
import bsf.baseservice.businessmq.producter.loadbalance.ErrorLoadBalancePartitionInfo;
import bsf.baseservice.businessmq.producter.loadbalance.SystemLoadBalance;
import bsf.baseservice.businessmq.systemruntime.log.ErrorLogHelper;
import bsf.baseservice.businessmq.systemruntime.*;
import bsf.db.DbConn;
import bsf.db.base.DbException;
import bsf.serialization.json.JsonProvider;

/**
 生产者提供者
 
*/
public class ProducterProvider implements AutoCloseable
{
	public BusinessMQConfig Config;
	public final String getMQPath()
	{
		return mqpath;
	}
	public final void setMQPath(String value)
	{
		mqpath = value.toLowerCase().trim();
	}
	public BaseLoadBalance LoadBalance = new SystemLoadBalance();
	public String ProducterName = "未定义";
	protected String mqpath;

	private ProducterContext context;
	private ProducterHeartbeatProtect producterHeartbeatProtect;
	private RedisNetCommand netCommand;

	public ProducterTimeWatchInfo ProducterTimeWatchInfo = new ProducterTimeWatchInfo();

	public ProducterProvider()
	{
		context = new ProducterContext();
		context.ProducterProvider=this;
		LoadBalance.Context=context;
	}

	public final void open()
	{
		try
		{
			DebugHelper.writeLine(-1, getMQPath(), "Open", "生产者开始初始化");
			//初始化上下文信息
			ProducterBLL bll = new ProducterBLL();
			try(DbConn c = DbConn.createConn(Config.ManageConnectString))
			{
				context.ProducterInfo=bll.getProducterInfo(c, getMQPath(), ProducterName);
				context.ProducterInfo.LoadBalance=LoadBalance;
				context.setManageServerTime(c.getServerDate());
			}
			context.IsNeedReload=false;
			context.LastMQPathUpdateTime=context.ProducterInfo.MqPathModel.lastupdatetime;

			//获取系统配置信息
			ConfigHelper.loadConfig(Config.ManageConnectString);

			netCommand = new RedisNetCommand(ConfigHelper.RedisServer);
			ProducterHeartbeatProtect.instance(context).Contexts.add(context); //注册上下文
			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Open", "生产者初始化成功");
			LogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Open", "生产者初始化成功");
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Open", "生产者初始化", exp);
			throw exp;
		}
	}
	/** 
	 发送消息
	 
	 <typeparam name="T"></typeparam>
	 @param objmsg
	*/
	public final <T> void sendMessage(T objmsg)
	{
		try
		{
			String json = "";
			//ProducterTimeWatchInfo.JsonHelperSerializer += TimeWatchLog.Debug(() =>
			//{
				if (!(objmsg instanceof String))
				{
					json = new JsonProvider().serialize(objmsg);
				}
					else
					{
						json = (String)((objmsg instanceof String) ? objmsg : null);
					}
			//}
		   //);
			//发送消息有n次重试机会
			int errortrycount = 0;
			while (errortrycount < context.ProducterInfo.LoadBalance.SendMessageErrorTryAgainCount)
			{
				LoadBalanceNodeInfo loadbalancenodeinfo = null;
				//ProducterTimeWatchInfo.GetLoadBalanceNodeInfo += Log.TimeWatchLog.Debug(() =>
				//{
					loadbalancenodeinfo = context.ProducterInfo.getLoadBalanceNodeInfo();
				//}
			   //);
				if (loadbalancenodeinfo == null)
				{
					throw new BusinessMQException(String.format("无法获取队列%1$s的可用的负载均衡数据节点", getMQPath()));
				}
				String datanodeconnectstring = new ProducterBLL().getDataNodeConnectString(SystemParamConfig.Producter_DataNode_ConnectString_Template_ToSendMessage, loadbalancenodeinfo.DataNodeModel);
				PartitionIDInfo partitionidinfo = PartitionRuleHelper.getPartitionIDInfo(loadbalancenodeinfo.MQPathPartitionModel.partitionid);
				java.util.Date manageservertime = context.getManageServerTime(); //.AddSeconds(1);发送消息要比标准时间提前1s，这样消息分表可以提前1s
				String tablename = PartitionRuleHelper.getTableName(partitionidinfo.TablePartition, manageservertime);
				try
				{
					//ProducterTimeWatchTest.AddMessages(string.Format("总插入消息:{0}s,插入消息:{1}s",allinserttime,inserttime));
					//ProducterTimeWatchInfo.SendMessage += Log.TimeWatchLog.Debug(() =>
					//{
						double inserttime = 0;
						//double allinserttime = Log.TimeWatchLog.Debug(() =>
						//{
							//inserttime = Log.TimeWatchLog.Debug(() =>
							//{
								tb_messagequeue_dal dal = new tb_messagequeue_dal();
								dal.TableName=tablename;

								try(DbConn c = DbConn.createConn(datanodeconnectstring))
								{
									tb_messagequeue_model tempVar = new tb_messagequeue_model();
									tempVar.message=json;
									tempVar.mqcreatetime=new java.util.Date();
									tempVar.sqlcreatetime=manageservertime;
									tempVar.source=(byte) EnumMessageSource.Common.Value;
									tempVar.state=(byte) EnumMessageState.CanRead.Value;
									dal.add2(c, tempVar);
								}
							//}
						   //);
						//}
					   //);
					//}
				   //);
					netCommand.sendMessage(mqpath);
					return;
				}
				catch (DbException exp)
				{
					ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "SendMessage", String.format("发送消息出现节点错误,节点:%1$s", loadbalancenodeinfo.DataNodeModel.datanodepartition), exp);
					context.ProducterInfo.removeMQPathPartition(loadbalancenodeinfo.DataNodeModel.datanodepartition); //数据层出错视为数据节点异常，则移除。将在一定时间内尝试恢复
					ErrorLoadBalancePartitionInfo t = new ErrorLoadBalancePartitionInfo();
					t.PartitionId=loadbalancenodeinfo.MQPathPartitionModel.partitionid;
					t.PartitionIndex=loadbalancenodeinfo.MQPathPartitionModel.partitionindex;
					context.ProducterInfo.LoadBalance.addError(t);
					//Context.IsNeedReload = true;
					if (context.SendMessageErrorTime == null)
					{
						context.SendMessageErrorTime=new java.util.Date();
					}

				}
				errortrycount++;
			}
			throw new BusinessMQException(String.format("发送消息出现系统级错误,并超过重试次数,请检查。队列:%1$s", getMQPath()));
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "SendMessage", "生产者发送消息出错", exp);
			throw new BusinessMQException("发送消息失败",exp);
		}
	}


	public final void close()
	{
		dispose();
	}

	/** 
	 资源释放
	 
	*/
	public final void dispose()
	{
		try
		{
			ProducterBLL bll = new ProducterBLL();
			try(DbConn c = DbConn.createConn(Config.ManageConnectString))
			{
				bll.removeProducter(c, context.ProducterInfo.ProducterModel.tempid, context.ProducterInfo.MqPathModel.id);
			}
			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者移除注册信息成功");
			LogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者移除注册信息成功");
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者移除注册信息出错", exp);
		}

		try
		{
			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者资源开始释放");
			ProducterHeartbeatProtect.instance(context).Contexts.remove(context); //移除上下文
			context.close(); //释放上下文

			DebugHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者资源释放成功");
			LogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者资源释放成功");
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(context.getMQPathID(), context.getMQPath(), "Dispose", "生产者资源释放出错", exp);
			throw exp;
		}

	}
}